package com.wudl.flink.examples;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @ClassName : test
 * @Description :
 * @Author :wudl
 * @Date: 2021-08-26 00:36
 */

public class test {
    public static void main(String[] args) {
        /* 流式table处理环境*/
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        /*创建读取binlog的表*/
        String sourceDDL = "CREATE TABLE my_binlog_cdc(" +
                "id INT NOT NULL," +
                "name STRING," +
                "age INT," +
                "PRIMARY KEY(id)  NOT ENFORCED" +
                ") WITH (" +
                " 'connector' = 'mysql-cdc'," +
                " 'hostname' = 'cdh101'," +
                " 'port' = '3306'," +
                " 'username' = 'root'," +
                " 'password' = 'nfdwdata'," +
                " 'database-name' = 'testdb'," +
                " 'table-name' = 'my_binlog_cdc'" +
                ")";

        /*写入kafka的表*/
        String sinkDDL = "CREATE TABLE binlog_sink(" +
                "id INT NOT NULL," +
                "name STRING," +
                "age INT" +
                ") WITH (" +
                " 'connector' = 'kafka'," +
                " 'topic' = 'luchangyin'," +
                " 'properties.bootstrap.servers' = 'cdh101:9092,cdh102:9092,cdh103:9092'," +
                " 'scan.startup.mode' = 'earliest-offset'," +
                " 'format' = 'debezium-json'" +
                ")";




        /*输入到控制台的表*/
//        String sinkDDL2 = "CREATE TABLE console_sink(" +
//                "id INT NOT NULL," +
//                "name STRING," +
//                "age INT" +
//                ") WITH (" +
//                " 'connector' = 'print'" +
//                ")";

        /*将binlog数据源写到kafka表*/
        String exeSql = "INSERT INTO binlog_sink SELECT * FROM my_binlog_cdc";

        /*将binlog数据源写到控制台表*/
//        String exeSql2 = "INSERT INTO console_sink SELECT * FROM my_binlog_cdc";

        tableEnv.executeSql(sourceDDL);
        tableEnv.executeSql(sinkDDL);
//        tableEnv.executeSql(sinkDDL2);

        tableEnv.executeSql(exeSql);
//        tableEnv.executeSql(exeSql2);
    }
}
